package cn.tedu.batch.broadcast

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration

import java.util

/**
 * @author Amos
 * @date 2022/5/22
 */

object BatchBroadcastDemo {
  def main(args: Array[String]): Unit = {
    // 创建批处理的环境
    val env = ExecutionEnvironment.getExecutionEnvironment

    // 构建两个数据集
    import org.apache.flink.api.scala._
    val scoreSource: DataSet[(Int, String, Int)] = env.fromCollection(List( (1, "语文", 50),(2, "数学", 70), (3, "英文", 86)))
    val studentSource: DataSet[(Int, String)] = env.fromCollection(List((1, "张三"), (2, "李四"), (3, "王五")))

    // 数据处理
    val result = scoreSource.map(new RichMapFunction[(Int, String, Int), (String, String, Int)] {
      // 定义全局变量
      var studentMap: Map[Int, String] = null


      // 只初始化一次
      override def open(parameters: Configuration): Unit = {
        // 获取广播变量，只需要获取一次即可
        import scala.collection.JavaConverters._
        val studentList: util.List[(Int, String)] = getRuntimeContext.getBroadcastVariable[(Int, String)]("student")
        // 将util.List转换为map,便于通过key找到value姓名的值
        studentMap = studentList.asScala.toMap
      }

      // (1, "语文", 50) => ("张三", "语文", 50)
      // 每来一条数据，调用一次map方法
      override def map(value: (Int, String, Int)): (String, String, Int) = {
        // 获取学生id
        val stuid = value._1
        val stuName = studentMap.getOrElse(stuid, "")
        (stuName, value._2, value._3)
      }
    }).withBroadcastSet(studentSource, "student")

    result.print()

  }

}
